package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.VideoPlay;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTrafficVideoPlayDetail {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // TODO 2 设置状态后端
        /*
            env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
            env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
            env.setStateBackend(new HashMapStateBackend());
            env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
            System.setProperty("HADOOP_USER_NAME", "atguigu");
        */

        //TODO 3 读取kafka播放主题数据
        String topicName = "dwd_play_video_log";
        String groupId = "dwd_traffic_video_play_detail";

        DataStreamSource<String> videoStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));

        SingleOutputStreamOperator<VideoPlay> mapStream = videoStream.map(new MapFunction<String, VideoPlay>() {
            @Override
            public VideoPlay map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Long ts = jsonObject.getLong("ts");
                String mid = jsonObject.getJSONObject("common").getString("mid");
                JSONObject appVideo = jsonObject.getJSONObject("appVideo");
                Long playSec = appVideo.getLong("play_sec");
                String videoId = appVideo.getString("video_id");
                return new VideoPlay(mid, videoId, playSec, ts);
            }
        });

        mapStream.print("json>>>");

        KeyedStream<VideoPlay, String> keyedStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<VideoPlay>forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner<VideoPlay>() {
            @Override
            public long extractTimestamp(VideoPlay element, long recordTimestamp) {
                return element.getTs();
            }
        })).keyBy(new KeySelector<VideoPlay, String>() {
            @Override
            public String getKey(VideoPlay videoPlay) throws Exception {
                return videoPlay.getMid();
            }
        });

        //TODO 5 开窗
        SingleOutputStreamOperator<VideoPlay> reduceStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L))).reduce(new ReduceFunction<VideoPlay>() {
            @Override
            public VideoPlay reduce(VideoPlay v1, VideoPlay v2) throws Exception {
                v1.setPlaySec(v1.getPlaySec() + v2.getPlaySec());
                return v1;
            }
        });


        reduceStream.print("reduce>>>");
        SingleOutputStreamOperator<String> result = reduceStream.map(new MapFunction<VideoPlay, String>() {
            @Override
            public String map(VideoPlay videoPlay) throws Exception {
                String s = JSONObject.toJSONString(videoPlay);
                return s;
            }
        });
        result.print("result>>>");

        result.addSink(KafkaUtil.getFlinkKafkaProducer("dwd_traffic_video_play_detail"));

        //TODO 6 执行任务
        env.execute(groupId);
    }
}
